package com.group2.edu.realtime.common.util;

import com.group2.edu.realtime.common.constant.Constant;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.serializer.SimpleStringSerializer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder;
import org.apache.flink.connector.kafka.sink.KafkaSink;

import java.util.Properties;

/**
 * @author 高耀
 * @date 2024/12/14 10:24<p></p>
 */
public class FlinkSinkUtil {
    /**
     * 此方法用于获取一个KafkaSink用于写往kafka一个固定的topic
     * @param topic 生产的topic主题
     */
    public static KafkaSink<String> getKafkaSink(String topic) {
        return KafkaSink.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS_ID)
                .setRecordSerializer(
                        new KafkaRecordSerializationSchemaBuilder<String>()
                                .setTopic(topic)
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
//                 .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) //精准一次
//                 .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60000 * 15 + "") //设置事务超时时间，大于保存点保存时间，小于kafka要求的最大事务超时时间15min
//                 .setTransactionalIdPrefix("tId" + System.currentTimeMillis()) //设置唯一事务id
                .build();
    }

    /**
     * 此方法用于用户自定义序列化器，
     * 可以根据数据自由决定向哪个topic中写入
     * @param serializationSchema 使用者自定义序列化器
     */
    public static <E> KafkaSink<E> getKafkaSink(KafkaRecordSerializationSchema<E> serializationSchema) {
        return KafkaSink.<E>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS_ID)
                .setRecordSerializer(serializationSchema)
//                 .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) //精准一次
//                 .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60000 * 15 + "") //设置事务超时时间，大于保存点保存时间，小于kafka要求的最大事务超时时间15min
//                 .setTransactionalIdPrefix("tId" + System.currentTimeMillis()) //设置唯一事务id
                .build();
    }

    public static DorisSink<String> getDorisSink(String tableName) {
        // 上游是 json 写入时，需要开启配置
        Properties properties = new Properties();
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");

        return DorisSink.<String>builder()
                .setDorisOptions(//普通设置项
                        DorisOptions.builder()
                                .setFenodes(Constant.DORIS_FE_NODES) //前端地址
                                .setTableIdentifier(Constant.DORIS_DATABASE + "." + tableName) //表名信息
                                .setUsername("root")
                                .setPassword("aaaaaa")
                                .build()
                )
                .setDorisExecutionOptions( //执行设置项
                        DorisExecutionOptions.builder()
                                .disable2PC()
                                .setDeletable(false)
                                .setStreamLoadProp(properties)
                                .build()
                )
                .setSerializer(new SimpleStringSerializer())//序列化器
                .build();
    }
}
